package per.chao.mqtt.domain.protocol.publish;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.facade.IMessageSender;
import per.chao.mqtt.domain.protocol.BaseProtocolHandler;
import per.chao.mqtt.domain.repo.PublishMessageRepo;
import per.chao.mqtt.domain.repo.RetainMessageRepo;
import per.chao.mqtt.domain.repo.SessionRepo;
import per.chao.mqtt.domain.repo.SubscribeMessageRepo;
import per.chao.mqtt.domain.shared.store.DupPublishMessageStore;
import per.chao.mqtt.domain.shared.store.RetainMessageStore;
import per.chao.mqtt.domain.shared.store.SubscribeStore;
import per.chao.mqtt.domain.vo.InternalMessage;

import java.util.List;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/19 23:05
 **/
@Component
@Slf4j
public class Publish extends BaseProtocolHandler {

	@Autowired
	@Qualifier("kafkaSender")
	private IMessageSender kafkaSender;

	@Autowired
	private RetainMessageRepo retainMessageRepo;
	@Autowired
	private SubscribeMessageRepo subscribeMessageRepo;
	@Autowired
	private SessionRepo sessionRepo;
	@Autowired
	private PublishMessageRepo publishMessageRepo;

	@Override
	public void handleMessage(Channel channel, Object msg) {
		MqttPublishMessage publishMessage = (MqttPublishMessage) msg;
		String clientId = (String) channel.attr(AttributeKey.valueOf("clientId")).get();

		log.info("{}", publishMessage.fixedHeader().qosLevel().value());

		// QoS=0
		if (publishMessage.fixedHeader().qosLevel() == MqttQoS.AT_MOST_ONCE) {
			byte[] msgBytes = new byte[publishMessage.payload().readableBytes()];
			publishMessage.payload().getBytes(publishMessage.payload().readerIndex(), msgBytes);
			InternalMessage internalMessage = InternalMessage.builder()
					.topic(publishMessage.variableHeader().topicName())
					.mqttQoS(publishMessage.fixedHeader().qosLevel().value())
					.messageBytes(msgBytes)
					.dup(false)
					.retain(false)
					.clientId(clientId)
					.build();
			kafkaSender.send(internalMessage);
			sendPublishMessage(publishMessage.variableHeader().topicName(), publishMessage.fixedHeader().qosLevel(),
					msgBytes, false, false);
		}

		// QoS=1
		if (publishMessage.fixedHeader().qosLevel() == MqttQoS.AT_LEAST_ONCE) {
			byte[] msgBytes = new byte[publishMessage.payload().readableBytes()];
			publishMessage.payload().getBytes(publishMessage.payload().readerIndex(), msgBytes);
			InternalMessage internalMessage = InternalMessage.builder()
					.topic(publishMessage.variableHeader().topicName())
					.mqttQoS(publishMessage.fixedHeader().qosLevel().value())
					.messageBytes(msgBytes)
					.dup(false)
					.retain(false)
					.clientId(clientId)
					.build();
			kafkaSender.send(internalMessage);
			sendPublishMessage(publishMessage.variableHeader().topicName(), publishMessage.fixedHeader().qosLevel(),
					msgBytes, false, false);
			sendPubAckMessage(channel, publishMessage.variableHeader().packetId());
		}

		// QoS=2
		if (publishMessage.fixedHeader().qosLevel() == MqttQoS.EXACTLY_ONCE) {
			byte[] msgBytes = new byte[publishMessage.payload().readableBytes()];
			publishMessage.payload().getBytes(publishMessage.payload().readerIndex(), msgBytes);
			InternalMessage internalMessage = InternalMessage.builder()
					.topic(publishMessage.variableHeader().topicName())
					.mqttQoS(publishMessage.fixedHeader().qosLevel().value())
					.messageBytes(msgBytes)
					.dup(false)
					.retain(false)
					.clientId(clientId)
					.build();
			kafkaSender.send(internalMessage);
			sendPublishMessage(publishMessage.variableHeader().topicName(), publishMessage.fixedHeader().qosLevel(), msgBytes, false, false);
			sendPubRecMessage(channel, publishMessage.variableHeader().packetId());
		}

		// retain=1，保留消息
		if (publishMessage.fixedHeader().isRetain()) {
			byte[] msgBytes = new byte[publishMessage.payload().readableBytes()];
			publishMessage.payload().getBytes(publishMessage.payload().readerIndex(), msgBytes);
			if (msgBytes.length == 0) {
				retainMessageRepo.remove(publishMessage.variableHeader().topicName());
			} else {
				RetainMessageStore retainMessageStore = new RetainMessageStore();
				retainMessageStore.setTopic(publishMessage.variableHeader().topicName());
				retainMessageStore.setMqttQoS(publishMessage.fixedHeader().qosLevel().value());
				retainMessageStore.setMessageBytes(msgBytes);
				retainMessageRepo.put(publishMessage.variableHeader().topicName(), retainMessageStore);
			}
		}
	}

	private void sendPubRecMessage(Channel channel, int packetId) {
		MqttMessage mqttPubRecMessage = MqttMessageFactory.newMessage(
				new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0),
				MqttMessageIdVariableHeader.from(packetId),
				null
		);
		channel.writeAndFlush(mqttPubRecMessage);
	}

	private void sendPubAckMessage(Channel channel, int packetId) {
		MqttPubAckMessage mqttPubAckMessage = (MqttPubAckMessage) MqttMessageFactory.newMessage(
				new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
				MqttMessageIdVariableHeader.from(packetId),
				null
		);
		channel.writeAndFlush(mqttPubAckMessage);
	}

	private void sendPublishMessage(String topicName, MqttQoS qosLevel, byte[] msgBytes, boolean retain, boolean dup) {
		List<SubscribeStore> subscribeStores = subscribeMessageRepo.getByTopic(topicName);
		for (SubscribeStore subscribeStore : subscribeStores) {
			MqttQoS respQoS = qosLevel.value() > subscribeStore.getMqttQoS() ? MqttQoS.valueOf(subscribeStore.getMqttQoS()) : qosLevel;
			// 判断当前订阅者是否在线
			if (sessionRepo.containsKey(subscribeStore.getClientId())) {
				if (respQoS == MqttQoS.AT_MOST_ONCE) {
					// QoS=0时最多发送一次，breaker不需要存储发布的消息
					// 订阅者收到mqtt消息的QoS级别，最终取决于发布消息的QoS和主题订阅的QoS
					MqttPublishMessage publishMsg = (MqttPublishMessage) MqttMessageFactory.newMessage(
							new MqttFixedHeader(MqttMessageType.PUBLISH, dup, respQoS, retain, 0),
							new MqttPublishVariableHeader(topicName, 0),
							Unpooled.buffer().writeBytes(msgBytes)
					);
					log.info("PUBLISH - clientId: {}, topic: {}, QoS: {}", subscribeStore.getClientId(), topicName, qosLevel.value());
					sessionRepo.get(subscribeStore.getClientId()).getChannel().writeAndFlush(publishMsg);
				}

				if (respQoS == MqttQoS.AT_LEAST_ONCE) {
					// QoS=1
					int messageId = 0; // next message id
					MqttPublishMessage publishMsg = (MqttPublishMessage) MqttMessageFactory.newMessage(
							new MqttFixedHeader(MqttMessageType.PUBLISH, dup, respQoS, retain, 0),
							new MqttPublishVariableHeader(topicName, messageId),
							Unpooled.buffer().writeBytes(msgBytes)
					);
					log.info("PUBLISH - clientId: {}, topic: {}, QoS: {}, messageId: {}", subscribeStore.getClientId(),
							topicName, respQoS.value(), messageId);
					DupPublishMessageStore dupPublishMessageStore = new DupPublishMessageStore();
					dupPublishMessageStore.setClientId(subscribeStore.getClientId());
					dupPublishMessageStore.setMessageBytes(msgBytes);
					dupPublishMessageStore.setMessageId(messageId);
					dupPublishMessageStore.setMqttQoS(respQoS.value());
					dupPublishMessageStore.setTopic(topicName);
					publishMessageRepo.put(subscribeStore.getClientId(), dupPublishMessageStore);
					sessionRepo.get(subscribeStore.getClientId()).getChannel().writeAndFlush(publishMsg);
				}

				if (respQoS == MqttQoS.EXACTLY_ONCE) {
					// QoS=2
					int messageId = 1; // next message id
					MqttPublishMessage publishMsg = (MqttPublishMessage) MqttMessageFactory.newMessage(
							new MqttFixedHeader(MqttMessageType.PUBLISH, dup, respQoS, retain, 0),
							new MqttPublishVariableHeader(topicName, messageId),
							Unpooled.buffer().writeBytes(msgBytes)
					);
					log.info("PUBLISH - clientId: {}, topic: {}, QoS: {}, messageId: {}", subscribeStore.getClientId(),
							topicName, respQoS.value(), messageId);
					DupPublishMessageStore dupPublishMessageStore = new DupPublishMessageStore();
					dupPublishMessageStore.setClientId(subscribeStore.getClientId());
					dupPublishMessageStore.setMessageBytes(msgBytes);
					dupPublishMessageStore.setMessageId(messageId);
					dupPublishMessageStore.setMqttQoS(respQoS.value());
					dupPublishMessageStore.setTopic(topicName);
					publishMessageRepo.put(subscribeStore.getClientId(), dupPublishMessageStore);
					sessionRepo.get(subscribeStore.getClientId()).getChannel().writeAndFlush(publishMsg);
				}
			}
		}
	}
}
